package com.test.business;

import com.alibaba.fastjson.JSON;
import com.test.entry.Order;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;

/**
 * 报警逻辑
 */
@Slf4j
public class FraudDetector extends KeyedProcessFunction<String, Order, String> {


    /**
     * ValueState 的作用域，只在当前的帐号中 accountNo
     */
    private transient ValueState<Boolean> flagState;

    private transient ValueState<Long> timeState;


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 初始化 ValueState
        ValueStateDescriptor<Boolean> descriptor = new ValueStateDescriptor("flag", Boolean.class);
        flagState = getRuntimeContext().getState(descriptor);

        // 初始化 ValueState
        ValueStateDescriptor<Long> timeFlag = new ValueStateDescriptor("time-Flag", Long.class);
        timeState = getRuntimeContext().getState(timeFlag);
    }

    /**
     * 每次订单流入，都会调用processElement 方法
     *
     * @param order
     * @param context
     * @param collector
     * @throws Exception
     */
    @Override
    public void processElement(Order order, Context context, Collector<String> collector) throws Exception {
        // 当帐号第一次进来/未设置 时， f=null
        Boolean f = flagState.value();
        if (f != null) {
            // f !=null ,该帐号上一笔订单<50，判断这笔金额是否>100
            if (order.getAmount().compareTo(new BigDecimal(100)) == 1) {
                // 危险帐号
                collector.collect(order.getAccountNo());
            }

            clearUp(context);

        } else {
            // 还未设置过帐号的状态，判断是否金额<50
            if (order.getAmount().compareTo(new BigDecimal(50)) == -1) {
                // 设置状态
                flagState.update(true);

                long time = context.timerService().currentProcessingTime() + 3000L;
                log.info("currentProcessingTime:{}", time);
                context.timerService().registerProcessingTimeTimer(time);
                timeState.update(time);
            }
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        flagState.clear();
        timeState.clear();
    }

    private void clearUp(Context context) throws Exception {
        // 清空帐号状态
        context.timerService().deleteEventTimeTimer(timeState.value());
        flagState.clear();
        timeState.clear();
    }
}
